package practice.library.gremlin_scala.trial.friends_group.script

import com.thinkaurelius.titan.core.schema.{PropertyKeyMaker, SchemaStatus, TitanManagement}
import com.thinkaurelius.titan.core.{Cardinality, TitanGraph}
import com.thinkaurelius.titan.graphdb.database.StandardTitanGraph
import com.thinkaurelius.titan.graphdb.database.management.{GraphIndexStatusReport, ManagementSystem}
import gremlin.scala.ScalaGraph
import org.apache.tinkerpop.gremlin.structure.{Direction, Edge}
import org.joda.time.{DateTime, Duration}
import utility.CustomLogger

import scala.util.{Failure, Success, Try}

/**
  * Created by lovepocky on 16/7/18.
  */
object BenchmarkTest extends CustomLogger {

    import practice.library.gremlin_scala.trial.friends_group.model.Vertex._
    import practice.library.gremlin_scala.trial.friends_group.model.Edge.label._
    import gremlin.scala._

    import org.apache.tinkerpop.gremlin.process.traversal.Order

    implicit def titanGraph2StandardTitanGraph(g: ScalaGraph[TitanGraph]): ScalaGraph[StandardTitanGraph] = g.asInstanceOf[ScalaGraph[StandardTitanGraph]]

    object TestEdgeLimit {

        import practice.library.gremlin_scala.trial.friends_group.script.Manager.Common

        object label {
            val normalNode = "normalNode"
            val partitionNode = "partitionNode"

            val normalEdge = "normalEdge"
        }

        object model {

            //vertex
            @label(label.normalNode)
            case class NodeCC(name: String)

            @label(label.partitionNode)
            case class PartitionNode(name: String)

            //edge
            @label(label.normalEdge)
            case class EdgeCC(count: Long, updateTime: DateTime, timeStamp: Long)

        }

        def normalGraphConfig(g: ScalaGraph[TitanGraph], addEdgeIndex: Boolean, addMixedIndex: Boolean) = {

            val mgmt = g.graph.openManagement()
            //build edge index
            //val vcIndexName = "vc_by_groupId_updateTime_of_edge"
            val vcIndexName = "vc_by_timeStamp_of_normalEdge"
            val esIndexName = "es_by_timeStamp_of_normalEdge"

            Common.closeAllTransactions(g)
            val timeStamp =
                if (mgmt.containsPropertyKey("timeStamp")) {
                    mgmt.getPropertyKey("timeStamp")
                } else {
                    Manager.PropertyKey.Create.timeStamp(mgmt) match {
                        case Success(s) => s
                        case Failure(e) => throw e
                    }
                }

            val edge =
                if (mgmt.containsEdgeLabel(label.normalEdge)) {
                    mgmt.getEdgeLabel(label.normalEdge)
                } else {
                    mgmt.makeEdgeLabel(label.normalEdge).make()
                }

            if (addEdgeIndex) {
                mgmt.buildEdgeIndex(edge, vcIndexName, Direction.BOTH, timeStamp)
            }

            if (addMixedIndex) {
                mgmt.buildIndex(esIndexName, classOf[Edge]).addKey(timeStamp).indexOnly(edge).buildMixedIndex("search")
            }
            mgmt.commit()

            if (addEdgeIndex) {
                ManagementSystem.awaitRelationIndexStatus(g.graph, vcIndexName, label.normalEdge).status(SchemaStatus.ENABLED).call()
            }

            if (addMixedIndex) {
                ManagementSystem.awaitGraphIndexStatus(g.graph, esIndexName).status(SchemaStatus.ENABLED).call()
            }

        }

        def partitionGraphConfig(g: ScalaGraph[TitanGraph]) = {
            val mgmt = g.graph.openManagement()
            mgmt.makeVertexLabel(label.partitionNode).partition().make()
            mgmt.commit()
        }

        def addVertex(g: ScalaGraph[TitanGraph], isPartitionVertex: Boolean, outVertexId: Option[Long] = None, inVertexId: Option[Long] = None, startNum: Int, endNum: Int) = {
            //bug fix: create node -> MUST commit -> get id
            val outVertex =
                if (isPartitionVertex) {
                    outVertexId.flatMap(g.V(_).headOption()).getOrElse(g + model.PartitionNode("out"))
                } else {
                    outVertexId.flatMap(g.V(_).headOption()).getOrElse(g + model.NodeCC("out"))
                }
            val inVertex =
                if (isPartitionVertex) {
                    outVertexId.flatMap(g.V(_).headOption()).getOrElse(g + model.PartitionNode("in"))
                } else {
                    outVertexId.flatMap(g.V(_).headOption()).getOrElse(g + model.NodeCC("in"))
                }
            loggers.console.info(s"outVertex: ${outVertex.id()}, inVertex: ${inVertex.id()}, isPartitionVertex: $isPartitionVertex")

            val step = 30000
            for (group <- Range(startNum, endNum, step)) {
                loggers.console.info(s"add edge start with $group")
                for (current <- Range(group, if (group + step > endNum) endNum else group + step)) {
                    val time = Generate.generateDateTime(DateTime.now().minusYears(1), DateTime.now())
                    outVertex --- model.EdgeCC(current, time, time.getMillis) --> inVertex
                }
                loggers.console.info("committing")
                g.tx().commit()
            }
        }

    }

    object TestVertexWithPointer {

        def addVertex(g: ScalaGraph[TitanGraph], friendsNum: Int, articleNum: Int, articleLength: Int, centerId: Option[Long] = None, startNum: Int = 0): String = {
            val limit = 50000
            if (articleNum < limit) {
                val step = limit / articleNum
                val centerVertex = centerId.map(g.V(_).head()).getOrElse(g + person("center"))
                val centerVertexId = centerVertex.id().toString.toLong
                loggers.console.info(s"centerVertexId: $centerVertexId")
                for (group <- Range(startNum, friendsNum + startNum, step)) {
                    for (num <- Range(group, if (group + step < friendsNum + startNum) group + step else friendsNum + startNum)) {
                        loggers.console.info(s"Creating Person $num")
                        val newFriend = g + person(s"friend_$num")
                        newFriend <-- friends --> centerVertex
                        //add article
                        for (articleCount <- Range(0, articleNum)) {
                            //generate new article
                            val time = Generate.generateDateTime(DateTime.now().minusYears(1), DateTime.now())
                            val newArticle = g + article(s"article_${num}_$articleCount", Generate.generateArticle(articleLength), time, time.getMillis, newFriend.id().toString.toLong)
                            newArticle --- writeBy --> newFriend

                            //generate new pointer
                            val pointer = g + pointer_article(centerVertexId, time.getMillis)
                            pointer --- pointer_readBy --> newArticle
                        }

                    }
                    loggers.console.info("committing")
                    g.tx().commit()
                }
                "complete"
            } else {
                "addVertex Stoped: articleNum too large"
            }
        }
    }

    //early version
    object TestVertex {

        //
        def addVertex(graph: ScalaGraph[TitanGraph], friendsNum: Int, articleNum: Int, articleLength: Int, centerId: Option[Long] = None): String /*center vertex id*/ = {
            val centerVertex = centerId.map(graph.V(_).head()).getOrElse(graph + person("center"))
            for (personNum <- 0 until friendsNum) {
                val newFriend = graph + person(s"friend_$personNum")
                loggers.console.info(s"new friends created: $personNum")
                newFriend <-- friends --> centerVertex
                for (articleNum <- 0 until articleNum) {
                    val time = Generate.generateDateTime(DateTime.now().minusYears(1), DateTime.now())
                    val newArticle = graph + article(s"article_${personNum}_$articleNum", Generate.generateArticle(articleLength), time, time.getMillis, newFriend.id().toString.toLong)
                    newArticle --- writeBy --> newFriend
                }
            }
            centerVertex.id().toString
        }

        def addVertex_withArticleIndexNode() = {

        }

        def deleteVertex(graph: ScalaGraph[TitanGraph], centerVertexId: Int) = {

        }

        //todo
        def findArticle(graph: ScalaGraph[TitanGraph], centerVertexId: Int, key: String, offset: Int, limit: Int): List[article] = {
            graph.V(centerVertexId).out(friends).in(writeBy).order().by(key, Order.decr).toList().map(_.toCC[article])

            /**
              * g.V(???).
              *
              */
        }

        def findArticle_OrderInScala(graph: ScalaGraph[TitanGraph], centerVertexId: Int, offset: Int, limit: Int): List[article] = {
            val start = DateTime.now
            val result = graph.V(centerVertexId).out(friends).in(writeBy)
                .toList().map(_.toCC[article]).sortWith((a, b) => a.updateTime.isAfter(b.updateTime)).slice(offset, limit)
            val end = DateTime.now()
            loggers.console.info(s"findArticle_OrderInScala take time: ${new Duration(start, end)}")
            result
        }
    }

    object Generate {

        //not perfect
        def generateDateTime(startTime: DateTime, endTime: DateTime): DateTime = {
            val newMillis = scala.math.abs(scala.util.Random.nextLong()) % (endTime.getMillis - startTime.getMillis) + startTime.getMillis
            new DateTime(newMillis)
        }

        def generateArticle(contentLength: Int) = {
            scala.util.Random.nextString(contentLength)
        }
    }

}
